//package test;
//
//import org.apache.flink.api.common.functions.FlatMapFunction;
//import org.apache.flink.api.common.functions.ReduceFunction;
//import org.apache.flink.api.java.utils.ParameterTool;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.util.Collector;
//
//
//public class flinkDemo {
//
//    public static void main(String[] args) throws Exception {
//        //连接端口号
//        final int port = 9000;
//        //获取执行环节
//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        //获取连接socket输入数据
//        DataStream<String> text = env.socketTextStream("node3.kg.cn", port, "\n");
//
//        //解析数据、对数据进行分组、窗口函数和统计个数
//
//        DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {
//
//                    private static final long serialVersionUID = 6800597108091365154L;
//
//                    public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
//                        for (String word : value.split("//s")) {
//                            out.collect(new WordWithCount(word, 1));
//                        }
//                    }
//                })
//                .keyBy("word")
//                .timeWindow(Time.seconds(5), Time.seconds(1))
//                .reduce(new ReduceFunction<WordWithCount>() {
//
//                    public WordWithCount reduce(WordWithCount value1, WordWithCount value2) throws Exception {
//
//                        return new WordWithCount(value1.word, value1.count + value2.count);
//                    }
//                });
//        windowCounts.print().setParallelism(1);
//
//        env.execute("Socket Window WordCount");
//
//    }
//
//    public static class WordWithCount {
//        public String word;
//        public long count;
//
//        public WordWithCount() {
//        }
//
//        public WordWithCount(String word, long count) {
//            this.word = word;
//            this.count = count;
//        }
//
//        @Override
//        public String toString() {
//            return word + " : " + count;
//        }
//    }
//}
//
